package com.roc.producer.messageType;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 请求回复消息-生产者
 *
 * @author Roker
 * @date 2025/07/08
 */
public class RequestReplyMsgProducer {
    private static final String REQUEST_TOPIC = "TOPIC_REQUEST";
    private static final String REPLY_TOPIC = "TOPIC_REPLY";
    private static final String GROUP_ID = "roc_producer_group";

    // 用于存储请求消息与对应的锁和条件变量，以等待应答消息
    private static final ConcurrentHashMap<String, RequestReplyInfo> requestMap = new ConcurrentHashMap<>();

    /**
     * 请求回复信息
     */
    static class RequestReplyInfo {
        private final Lock lock = new ReentrantLock();// 用于等待应答消息，默认情况下，锁和条件变量都是非公平锁
        private final Condition condition = lock.newCondition();// 用于等待应答消息，默认情况下，锁和条件变量都是非公平锁
        private String replyMessage;// 应答消息内容
        private boolean isReplyReceived = false;// 是否收到应答消息，默认为false
    }

    public static void main(String[] args) throws Exception {
        // 初始化生产者
        DefaultMQProducer producer = new DefaultMQProducer(GROUP_ID);
        producer.setNamesrvAddr("localhost:9876");
        producer.start();

        // 初始化消费者，用于接收应答消息
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(GROUP_ID + "_reply_consumer");
        consumer.setNamesrvAddr("localhost:9876");
        consumer.subscribe(REPLY_TOPIC, "*");

        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    try {
                        String requestId = msg.getUserProperty("requestId");
                        String replyMessage = new String(msg.getBody(), "UTF-8");
                        RequestReplyInfo info = requestMap.get(requestId);
                        if (info != null) {
                            info.lock.lock();
                            try {
                                info.replyMessage = replyMessage;
                                info.isReplyReceived = true;
                                info.condition.signal();
                            } finally {
                                info.lock.unlock();
                            }
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        consumer.start();

        // 发送请求消息
        for (int i = 0; i < 3; i++) {
            String requestId = UUID.randomUUID().toString();
            Message requestMessage = new Message(REQUEST_TOPIC, ("请求消息内容 - " + i).getBytes());
            requestMessage.putUserProperty("requestId", requestId);

            RequestReplyInfo info = new RequestReplyInfo();
            requestMap.put(requestId, info);

            producer.send(requestMessage);

            // 等待接收应答消息，设置超时时间为5秒
            info.lock.lock();
            try {
                if (!info.isReplyReceived) {
                    info.condition.await(5, TimeUnit.SECONDS);
                }
                if (info.isReplyReceived) {
                    System.out.println("收到应答消息：" + info.replyMessage);
                } else {
                    System.out.println("等待应答消息超时");
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                info.lock.unlock();
                requestMap.remove(requestId);
            }
        }

        producer.shutdown();
        consumer.shutdown();
    }
}